package activemq;

import javax.jms.Connection;
import javax.jms.ConnectionFactory;
import javax.jms.Destination;
import javax.jms.JMSException;
import javax.jms.Message;
import javax.jms.MessageConsumer;
import javax.jms.MessageListener;
import javax.jms.MessageProducer;
import javax.jms.Session;

import org.apache.activemq.ActiveMQConnectionFactory;

public class Consumer {
	private ConnectionFactory connectionFactory = null;
	private Connection connection = null;
	private Session session = null;

	public Connection getConnection() {
		return connection;
	}

	public void setConnection(Connection connection) {
		this.connection = connection;
	}

	public Session getSession() {
		return session;
	}

	public void setSession(Session session) {
		this.session = session;
	}

	public Consumer() {
		init();
	}

	private void init() {
		connectionFactory = new ActiveMQConnectionFactory(ActiveMQConnectionFactory.DEFAULT_USER,
				ActiveMQConnectionFactory.DEFAULT_PASSWORD, "tcp://0.0.0.0:6666");
		Connection connection = null;
		try {
			connection = connectionFactory.createConnection();
			connection.start();
			session = connection.createSession(false, Session.CLIENT_ACKNOWLEDGE);
		} catch (Exception e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
			try {
				connection.close();
			} catch (JMSException e1) {
				// TODO Auto-generated catch block
				e1.printStackTrace();
			}
		}
	}
	
	public static void runSimpleConsumer(){
		Consumer consumer = new Consumer();
		Session session = consumer.getSession();
		Destination dest;
		try {
			dest = session.createQueue("zjj");
			MessageConsumer msgConsumer = session.createConsumer(dest);
			while (true) {	
				Thread.sleep(10000);
				Message msg = msgConsumer.receive();
				System.out.println(msg.getStringProperty("zjj"));
				msg.acknowledge();
			}
		} catch (Exception e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		} finally {
			if (consumer.getConnection() != null) {
				try {
					consumer.getConnection().close();
				} catch (JMSException e1) {
					// TODO Auto-generated catch block
					e1.printStackTrace();
				}
			}

		}
	}
	
	public static void runConditionConsumer(){
		Consumer consumer = new Consumer();
		Session session = consumer.getSession();
		Destination dest;
		try {
			dest = session.createQueue("condition");
			String condition="number%2=1";
			MessageConsumer msgConsumer = session.createConsumer(dest,condition);							
			msgConsumer.setMessageListener(new MessageListener() {					
				public void onMessage(Message msg) {					
					try {
						Thread.sleep(10000);
						System.out.println(msg.getStringProperty("number"));
						msg.acknowledge();
					} catch (Exception e) {
						// TODO Auto-generated catch block
						e.printStackTrace();
					}					
				}
			});			
		} catch (Exception e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		} finally {
			if (consumer.getConnection() != null) {
				try {
					consumer.getConnection().close();
				} catch (JMSException e1) {
					// TODO Auto-generated catch block
					e1.printStackTrace();
				}
			}

		}
	}

	public static void main(String[] args) {
		runConditionConsumer();
	}
}
